package cn.doitedu.rtdw.dataetl;

import cn.doitedu.rtdw.beans.AccessTimeLongBean;
import cn.doitedu.rtdw.udfs.MyTimeTruncUdf;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

public class JOB03_AccessTimeLongOLAP_Support {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 读取kafka中dwd明细宽表数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("dwd-events")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("doit42-etl-job03-t666")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> s1 = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        SingleOutputStreamOperator<AccessTimeLongBean> beanStream = s1.map(new RichMapFunction<String, AccessTimeLongBean>() {
            AccessTimeLongBean accessTimeLongBean;

            @Override
            public void open(Configuration parameters) throws Exception {
                accessTimeLongBean = new AccessTimeLongBean();
            }

            @Override
            public AccessTimeLongBean map(String logJson) throws Exception {

                JSONObject jsonObject = JSON.parseObject(logJson);
                Long userId = jsonObject.getLong("user_id");
                String sessionId = jsonObject.getString("session_id");
                JSONObject properties = jsonObject.getJSONObject("properties");
                String url = properties.getString("url");
                String pageType = jsonObject.getString("page_type");
                String deviceType = jsonObject.getString("device_type");
                String releaseChannel = jsonObject.getString("release_channel");
                Long actionTime = jsonObject.getLong("action_time");
                String eventId = jsonObject.getString("event_id");

                accessTimeLongBean.set(userId, sessionId, url, pageType, deviceType, releaseChannel,actionTime,eventId);
                return accessTimeLongBean;
            }
        });

        KeyedStream<AccessTimeLongBean, String> keyedStream = beanStream.keyBy(bean -> bean.getSession_id());

        // 虚拟事件插值
        SingleOutputStreamOperator<AccessTimeLongBean> czStream = keyedStream.process(new KeyedProcessFunction<String, AccessTimeLongBean, AccessTimeLongBean>() {

            ValueState<AccessTimeLongBean> beanState;

            @Override
            public void open(Configuration parameters) throws Exception {

                beanState = getRuntimeContext().getState(new ValueStateDescriptor<AccessTimeLongBean>("beanState", AccessTimeLongBean.class));

            }

            @Override
            public void processElement(AccessTimeLongBean logBean, KeyedProcessFunction<String, AccessTimeLongBean, AccessTimeLongBean>.Context ctx, Collector<AccessTimeLongBean> out) throws Exception {

                // 判断状态是否为空，为空则先把第一条数据塞进去
                if (beanState.value() == null) {
                    logBean.setPage_start_time(logBean.getAction_time());
                    beanState.update(logBean);
                }


                /* pageload事件：状态中的上一条数据（代表上一个页面）的 action_time 更新,输出;然后本条数据放入状态，并再次输出  */
                if ("page_load".equals(logBean.getEvent_id())) {
                    // 更新掉状态中之前数据的 action_time，并输出
                    // 为了封闭掉前一个页面的结束
                    AccessTimeLongBean stateBean = beanState.value();
                    stateBean.setAction_time(logBean.getAction_time());
                    out.collect(stateBean);

                    // 补全新页面起始时间 page_start_time，放回状态，并输出，作为新页面的第一条数据
                    logBean.setPage_start_time(logBean.getAction_time());
                    beanState.update(logBean);
                    out.collect(logBean);
                }

                // 收到 wakeup事件
                else if ("wake_up".equals(logBean.getEvent_id())) {
                    // 补全新的页面起始时间
                    logBean.setPage_start_time(logBean.getAction_time());
                    // 覆盖状态
                    beanState.update(logBean);
                    // 将自己输出
                    out.collect(logBean);

                }


                /* 其他事件：将状态中的上一条数据的 action_time 更新，并输出  */
                else {
                    AccessTimeLongBean stateBean = beanState.value();
                    // 更新状态中的stateBean的action_time
                    stateBean.setAction_time(logBean.getAction_time());
                    // 然后将更新后的状态数据输出
                    out.collect(stateBean);

                }

            }
        });

        // 流转表
        tenv.createTemporaryView("cz",czStream,
                Schema.newBuilder()
                        .column("user_id", DataTypes.BIGINT())
                        .column("session_id",DataTypes.STRING())
                        .column("url",DataTypes.STRING())
                        .column("page_type",DataTypes.STRING())
                        .column("device_type",DataTypes.STRING())
                        .column("release_channel",DataTypes.STRING())
                        .column("action_time",DataTypes.BIGINT())
                        .column("event_id",DataTypes.STRING())
                        .column("page_start_time",DataTypes.BIGINT())
                        .columnByExpression("rt","to_timestamp_ltz(action_time,3)")
                        .watermark("rt","rt - interval '0' second ")
                .build());


        // 创建doris的逻辑映射表
        tenv.executeSql(
                " CREATE  TABLE  access_timelong_agg_doris("+
                        "   dt   date,                           "+
                        "   user_id bigint,                      "+
                        " 	session_id varchar(20),              "+
                        " 	url varchar(50),                     "+
                        " 	page_type varchar(20),               "+
                        " 	device_type varchar(20),             "+
                        " 	window_start_hour  varchar(40),      "+
                        " 	release_channel varchar(20),         "+
                        " 	page_timelong bigint                 "+
                        " ) WITH (                                             "+
                        "    'connector' = 'doris',                            "+
                        "    'fenodes' = 'doitedu:8030',                       "+
                        "    'table.identifier' = 'dws.access_timelong_agg',   "+
                        "    'username' = 'root',                              "+
                        "    'password' = 'root',                              "+
                        "    'sink.label-prefix' = 'doit42-05'                 "+
                        " )                                                    "
        );

        tenv.createTemporaryFunction("my_time_trunc", MyTimeTruncUdf.class);
        tenv.executeSql(
                "INSERT INTO access_timelong_agg_doris  " +
                        "SELECT                                                                                                                  "+
                        "   to_date(date_format(window_start,'yyyy-MM-dd')) as dt,                                                              "+
                        "   user_id,                                                                                                            "+
                        " 	session_id,                                                                                                           "+
                        " 	url,                                                                                                                  "+
                        " 	page_start_time,                                                                                                                  "+
                        " 	page_type,                                                                                                            "+
                        " 	device_type,                                                                                                          "+
                        " 	my_time_trunc(unix_timestamp(date_format(window_start,'yyyy-MM-dd HH:mm:ss.SSS'))*1000,60) AS window_start_hour,       "+
                        " 	release_channel,                                                                                                      "+
                        " 	max(action_time - page_start_time) as page_timelong                                                                  "+
                        " FROM TABLE(                                                                                                             "+
                        "     TUMBLE(TABLE cz,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                                                                 "+
                        " )                                                                                                                       "+
                        " GROUP BY                                                                                                                "+
                        "     window_start,                                                                                                       "+
                        " 	window_end,                                                                                                           "+
                        "     user_id,                                                                                                            "+
                        " 	session_id,                                                                                                           "+
                        " 	url,                                                                                                                  "+
                        " 	page_type,                                                                                                            "+
                        " 	device_type,                                                                                                          "+
                        " 	release_channel,                                                                                                      "+
                        " 	page_start_time                                                                                                       "
        );

        env.execute();
    }
}
